package org.zn.note.jdk;

import org.zn.note.common.util.ThreadUtil;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.*;

/*
乐观锁/悲观锁：不是具体的实现，而是从并发同步角度出发的一种概念或思想。
    场景：存款时，取出余额，然后将余额+存款额，然后更新余额。
    乐观锁(多读，其实就是不加锁)：
       1、对数据有个version
        ① select 余额, version  ② 余额++  ③ update 余额, version+1 where version
       2、CAS（compareAndSet）（相关实现AtomicInteger）
        内存V，预期值A，修改值B：只有$V==A，才$V=B。保证一个变量修改的原子性，如AtomicInteger。
        缺点 ① 不能保证一个代码块的原子性  ② 要不断轮询，CPU消耗大（自旋）  ③ 有ABA问题，就是值被改了又被改回来了
    悲观锁(多写，就是加锁，可能有读锁写锁等等)：
        从取余额开始，就加锁，不让别人读写，直到更新后释放。
        因为这个过程是要修改余额的，怕别人读到不对的数据，也怕别人写了数据自己处理的不对了。
独享锁(X锁/互斥锁/独占锁/写锁)：事物T对数据A加了X锁，则其他事物不能加任何锁。
共享锁(S锁/读锁)：事物T对数据A加了X锁，则其他事物也只能加X锁。只能读，不能写。
可重入锁：事物T获取了对数据A的锁，没有释放，这时事物T再次获取对数据A的锁，还是能够获取到的。（同一事务重复获取一把锁）。
公平锁/非公平锁：是否按照排队顺序分配锁。

累加【50(线程)*1KW(次累加)】：
① 非线程安全几十毫秒(累加的不对，没意义)
② 加锁，2W多毫秒
③ AtomicInteger，1W多毫秒

AtomicInteger 实现原理
JVM提供了一种CAS的原子实现，compareAndSwapInt(内存地址，预期值，修改值)
do {
   取预期值;
   修改值 = 预期值 + 1
} while(compareAndSwapInt(内存地址，预期值，修改值)) // 如果false，在重新取当前的预期值，重新计算，自旋

synchronized
public synchronized void method1(){}  当前对象实例
public static synchronized void method1(){}  当前类的class对象
synchronized(object) {}  object对象

Java AQS (AbstractQueuedSynchronizer)  阻塞锁 + FIFO等待队列同步器
1、维护了一个 volatile int state，表示用户同步状态，提供了原子操作（unsafe.compareAndSwapInt实现）
   getState()、setState()、compareAndSetState()
2、提供了互斥模式(exclusive)、共享模式(shared)，一般实现其中一种，也有实现两种的
   互斥：ReentrantLock  共享：Semaphore/CountDownLatch  两种：ReadWriteLock
 */
public class JavaLock {

    private static int count1 = 0;
    private static int count2 = 0;
    private static AtomicInteger count3 = new AtomicInteger(0);
    private static AtomicReference count4 = new AtomicReference();

    /*
    ReentrantLock 可重入独占锁
    两个构造方法：① 默认非公平锁  ② ReentrantLock(boolean fair)
    公平锁：FairSync extends Sync    非公平锁 NonfairSync extends Sync
    Sync extends AbstractQueuedSynchronizer --> AQS
     */
    private static void testReentrantLock() {
        ReentrantLock reentrantLock = new ReentrantLock();
        try {
            reentrantLock.lock(); // 获取不到阻塞
            reentrantLock.tryLock(); // 获取不到false
            reentrantLock.tryLock(10L, TimeUnit.SECONDS); // 有超时的阻塞
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != reentrantLock) {
                reentrantLock.unlock();
            }
        }
    }

    /*
    ReentrantReadWriteLock
    也是两个构造函数：① 默认非公平锁  ② ReentrantReadWriteLock(boolean fair)
    内有两把锁：
    public interface ReadWriteLock {
        Lock readLock(); // 共享锁
        Lock writeLock(); // 独占锁
    }
     */
    private static void testReentrantReadWriteLock() {
        ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        reentrantReadWriteLock.writeLock().lock();
        reentrantReadWriteLock.readLock().lock();
    }

    /*
    AQS
    AbstractQueuedSynchronizer （AQS 抽象排队同步器）  构建锁和同步器的框架
    比如：ReentrantLock、Semaphore、ReentrantReadWriteLock、SynchronousQueue、FutureTask
    核心思想：
    1、如果共享资源空闲，设置当前请求线程为有效线程，并锁定共享资源。
    2、如果共享资源锁定，则需要一套线程阻塞以及唤醒时锁分配的机制。AQS用CLH队列锁实现的。
    AQS将每个请求线程封装成一个CHL锁的节点。

    CHL（Craig.Landin 和 Hagersten 发明的 Lock），虚拟双向队列（不存在队列实例，仅存在节点之间的关联关系）。
    通过一定手段将所有线程对某一共享变量轮询竞争转化为一个线程队列且队列中的线程各自轮询自己的本地变量。

    AQS资源共享方式：
    1、Exclusive：如 ReentrantLock，又分公平和非公平
    2、Share：如 Semaphore/CountDownLatch

    自定义的同步器，只需要实现共享资源state的获取和释放方式。线程等待队列AQS实现了。（模板模式）

    模板模式：(基于继承，而策略模式基于组合)
        好莱坞原则(别打给我，我会打给你) -->
        上层（超类）控制流程，下层（子类）只需要实现抽象方法，不用调用上层的方法，上层会“自动”调用下层实现的方法。

    AQS提供的模板方法：（其他方法都是final的，“流程性”的，不让动）
    isHeldExclusively()//该线程是否正在独占资源。只有用到condition才需要去实现它。
    tryAcquire(int)//独占方式。尝试获取资源，成功则返回true，失败则返回false。
    tryRelease(int)//独占方式。尝试释放资源，成功则返回true，失败则返回false。
    tryAcquireShared(int)//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
    tryReleaseShared(int)//共享方式。尝试释放资源，成功则返回true，失败则返回false。

    ReentrantLock
      abstract static class Sync extends AbstractQueuedSynchronizer {}
        static final class FairSync extends Sync {} // 公平
        static final class NonfairSync extends Sync {} // 非公平
    1、首先获取锁这块，就是看能否 CAS-state(0-->1)，哪个线程成功把 state从0变成1了，就代表获取到锁了
       可重入的，该线程每lock一次，state+1；最终要unlock N次，才能释放掉！
    2、加到队列里，然后每个线程各自循环尝试步骤1，起到阻塞作用
     */
    private static void testAbstractQueuedSynchronizer() {
        // 用 ReentrantLock 看一下
        Lock lock = new ReentrantLock();
        try {
            lock.lock();
        } finally {
            lock.unlock();
        }
    }

    /*
    Semaphore
    构造1：Semaphore(int) 默认非公平    构造2：Semaphore(int permits, boolean fair)
    用于限制获取某种资源的线程数

    AQS共享模式，用的 tryAcquireShared(int)。  （上面）
     */
    private static void testSemaphore() { // 信号
        int permits = 5; // 5个许可
        Semaphore semaphore = new Semaphore(permits);
        try {
            semaphore.acquire(); // 获取一个许可
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release(); // 释放一个许可
        }
    }

    /*
    Condition (java.util.concurrent.locks) “条件”
    对应关系
    Object.wait()           condition.await()         释放锁
    Object.notify()         condition.signal()        不释放锁
    Object.notifyAll()      condition.signalAll()      不释放锁
    和Object一样，condition也需要获取到对应的lock才能用

    区别：
    从应用层面上，一个Lock可以new多个Condition，这样 await和signal更灵活。 而Object monitor的notify不一定通知到哪个wait。

    Condition原理：条件队列，当一个线程在调用了await方法以后，直到线程等待的某个条件为真的时候才会被唤醒。
     */
    private static void testCondition() {
        final Lock lock = new ReentrantLock();
        //AQS里有个实现：AbstractQueuedSynchronizer.ConditionObject
        final Condition condition = lock.newCondition();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    ThreadUtil.printThread("start ...");
                    lock.lock();
                    ThreadUtil.printThread("111");
                    condition.await();
                    ThreadUtil.printThread("222");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                    ThreadUtil.printThread("end ...");
                }
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    ThreadUtil.printThread("start ...");
                    lock.lock();
                    condition.signal();
                    ThreadUtil.sleep(1L);
                    ThreadUtil.printThread("111");
                } finally {
                    lock.unlock();
                    ThreadUtil.printThread("end ...");
                }
            }
        });
        thread1.start();
        ThreadUtil.sleep(1L);
        thread2.start();
    }

    private static void testCyclicBarrier() {
        int threadNum = 5;
        final CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum);
        new Thread(new Runnable() {
            @Override
            public void run() {
                ThreadUtil.printThread("start ...");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public static void main(String[] args) throws InterruptedException {
        testCondition();
    }

    private static void test() throws InterruptedException {
        int taskExecutorNumbers = 50;
        final CountDownLatch countDownLatch = new CountDownLatch(taskExecutorNumbers);
        ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(taskExecutorNumbers);
        executorService.prestartAllCoreThreads();
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < taskExecutorNumbers; i++)
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 10000000; i++) {
                        // 1、非同步的，线程不安全，count累计错误
//                        count1++;
                        // 2、加锁（阻塞同步）
//                        synchronized (this.getClass()) {
//                            count2++;
//                        }
//                        // 3、用原子类（非阻塞同步）
                        count3.incrementAndGet();
                    }
                    countDownLatch.countDown();
                }
            });
        executorService.shutdown();
        countDownLatch.await();
        long endTime = System.currentTimeMillis();
        System.out.printf("%d, %d, %d, Time[%d]", count1, count2, count3.intValue(), endTime - startTime);
    }
}
